package com.atguigu.tingshu.user.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.MongoUtil;
import com.atguigu.tingshu.model.user.UserListenProcess;
import com.atguigu.tingshu.user.service.UserListenProcessService;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.atguigu.tingshu.vo.user.UserListenProcessVo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.print.attribute.standard.PrinterURI;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
@SuppressWarnings({"all"})
public class UserListenProcessServiceImpl implements UserListenProcessService {

    @Autowired
    private MongoTemplate mongoTemplate;

    /**
     * 查询当前登录用户对指定声音播放进度
     *
     * @param userId
     * @param trackId
     * @return
     */
    @Override
    public BigDecimal getTrackBreakSecond(Long userId, Long trackId) {
        //1.获取用户播放进度集合名称
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
        //2.构建查询对象 查询条件：用户ID+声音ID
        Query query = new Query();
        query.addCriteria(Criteria.where("userId").is(userId).where("trackId").is(trackId));
        UserListenProcess userListenProcess = mongoTemplate.findOne(query, UserListenProcess.class, collectionName);
        if (userListenProcess != null) {
            return userListenProcess.getBreakSecond();
        }
        return new BigDecimal("0.00");
    }

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private KafkaService kafkaService;

    /**
     * 保存/更新当前登录用户对指定声音播放进度;更新声音及所属专辑统计信息（播放量）
     *
     * @param userListenProcessVo
     * @return
     */
    @Override
    public void updateListenProcess(Long userId, UserListenProcessVo userListenProcessVo) {
        //1.获取用户播放进度集合名称
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
        //2.构建查询对象 查询条件：用户ID+声音ID
        Query query = new Query();
        Long trackId = userListenProcessVo.getTrackId();
        query.addCriteria(Criteria.where("userId").is(userId).where("trackId").is(trackId));
        UserListenProcess userListenProcess = mongoTemplate.findOne(query, UserListenProcess.class, collectionName);
        //3.如果播放进度对象为空则新增播放进度
        userListenProcessVo.setBreakSecond(userListenProcessVo.getBreakSecond().setScale(2, RoundingMode.HALF_UP));
        if (userListenProcess == null) {
            //3.1 拷贝VO中专辑ID，声音ID，播放进度
            userListenProcess = BeanUtil.copyProperties(userListenProcessVo, UserListenProcess.class);
            userListenProcess.setUserId(userId);
            userListenProcess.setCreateTime(new Date());
            userListenProcess.setUpdateTime(new Date());
        } else {
            //4.如果播放进度对象有值则更新播放进度时间及更新时间
            userListenProcess.setUpdateTime(new Date());
            userListenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
        }
        mongoTemplate.save(userListenProcess, collectionName);

        //5.TODO 更新统计信息（MySQL,ElasticSearch） 业务要求：当天内对于统计数值限制最多增量+1
        //5.1 采用redis String提供命令：set k v ex nx 实现
        //5.1.1 构建key的值 前缀:用户ID_专辑ID_声音ID
        Long albumId = userListenProcessVo.getAlbumId();
        String key = RedisConstant.USER_TRACK_REPEAT_STAT_PREFIX + userId + "_" + albumId + "_" + trackId;
        //5.1.2 计算过期时间：当天结束时间-当前系统时间
        DateTime endofDayDateTime = DateUtil.endOfDay(new Date());
        long ttl = endofDayDateTime.getTime() - DateUtil.currentSeconds();

        //5.1.3 将Key尝试存入Redis
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, null, ttl, TimeUnit.MILLISECONDS);
        //5.2 setnx成功，构建更新统计MQ对象，发送消息到指定话题
        if (flag) {
            //5.2.1  构建更新统计MQ对象
            TrackStatMqVo mqVo = new TrackStatMqVo();
            mqVo.setAlbumId(albumId);
            mqVo.setTrackId(trackId);
            mqVo.setStatType(SystemConstant.TRACK_STAT_PLAY_NAME);
            mqVo.setCount(1);
            //5.2.2 构建MQ消息唯一编号，在消费者端（专辑服务，搜索服务）确保消息幂等性
            mqVo.setBusinessNo(IdUtil.fastSimpleUUID());
            //5.2.2 发送MQ消息到话题
            kafkaService.sendMessage(KafkaConstant.QUEUE_TRACK_STAT_UPDATE, JSON.toJSONString(mqVo));
        }

    }

    /**
     * 获取当前用户上次播放声音信息
     *
     * @return
     */
    @Override
    public Map<String, Long> getLatelyTrack(Long userId) {
        //1.获取用户播放进度集合名称
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
        //2.构建查询对象：查询条件-用户ID，排序字段-更新时间，限制返回一条记录
        Query query = new Query(Criteria.where("userId").is(userId));
        query.with(Sort.by(Sort.Direction.DESC, "updateTime"));
        UserListenProcess userListenProcess = mongoTemplate.findOne(query, UserListenProcess.class, collectionName);
        if(userListenProcess!=null){
            Map<String, Long> map = new HashMap<>();
            map.put("albumId", userListenProcess.getAlbumId());
            map.put("trackId", userListenProcess.getTrackId());
            return map;
        }
        return null;
    }
}
